package com.algo.panelmn;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;  
import org.apache.hadoop.conf.Configured;  
import org.apache.hadoop.fs.Path;  
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.Tool;  
import org.apache.hadoop.util.ToolRunner;

import com.algo.panelmn.Method2CookieUVStep1.Mappersns;
  
public class Method2CookieUVStep2 extends Configured implements Tool {
	
	//get the snsid mzid mapping in the campaign time
	public static class Mappersns extends org.apache.hadoop.mapreduce.Mapper<LongWritable, Text, Text, Text> {
		public void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String str = value.toString();
			String[] sns = str.split("\\^");
			if(sns.length >= 4){
				String uuid = sns[0];
				String snsid = sns[1];
				context.write(new Text(snsid), new Text("uuid-" + uuid));
			}
			else	System.err.println("less sns element");
		}
	}
	//read (mzid ,snsid) mapping generated by step1
	public static class Mappercamp extends org.apache.hadoop.mapreduce.Mapper<Text, Text, Text, Text>{
		public void map(Text key, Text value, Context context)
				throws IOException, InterruptedException {
			context.write(value, key);
		}
	}
	public static class Reducer extends org.apache.hadoop.mapreduce.Reducer<Text, Text, Text, LongWritable>{
		//the same snsid was passed to the same reducer
		//output mzid and its proportion in snsid
		public void reduce(Text key, Iterable<Text> values,Context context)
			 throws IOException, InterruptedException {
			Set uuids = new HashSet();
			Set campuuids = new HashSet();
			Iterator it = values.iterator();
			while(it.hasNext()){
				String lw = ((Text)it.next()).toString();
				if(lw.indexOf("uuid-") == 0){
					lw = lw.substring(5);
					if(!uuids.contains(lw) && !campuuids.contains(lw))	uuids.add(lw);
				}
				else{
					if(!campuuids.contains(lw))	campuuids.add(lw);
					if(uuids.contains(lw))	uuids.remove(lw);
				}
			}
			long weight = campuuids.size() + uuids.size();
			it = campuuids.iterator();
			while(it.hasNext()){
				String uuid = (String) it.next();
				context.write(new Text(uuid), new LongWritable(weight));
			}
		}
	}
	public int getMonthDay(int year, int month){
		if(month > 12 || month < 1) return 0;
		if(month%2 == 1 && month <= 7) return 31;
		else if(month%2 == 0 && month >= 8) return 31;
		else if(month != 2) return 30;
		else{
			if((year%4 == 0 && year%100 != 0) || year%400 == 0) return 29;
			else return 28;
		}
	}
    public int run(String[] args) throws Exception {
    	for(int i = 0; i< args.length; i++)
			System.out.println("step2:" + args[i]);
        Configuration conf = getConf();
        Job job = new Job(conf, "cookie-step2");
        job.setJarByClass(Method2CookieUVStep2.class);
        
        int reducernum = 0;
        Path p1 = new Path(args[0]);
        MultipleInputs.addInputPath(job, p1, KeyValueTextInputFormat.class, Mappercamp.class);
        //MultipleInputs.addInputPath(job, p1, LzoTextInputFormat.class, Mappercamp.class);
        int fromyear = Integer.parseInt(args[2].substring(0, 4));
        int toyear = Integer.parseInt(args[3].substring(0, 4));
        int frommonth = Integer.parseInt(args[2].substring(4, 6));
        int tomonth = Integer.parseInt(args[3].substring(4, 6));
        int fromday = Integer.parseInt(args[2].substring(6));
        int today = Integer.parseInt(args[3].substring(6));
        for(int i = fromyear; i <= toyear; i++){
		int tomonth_y = tomonth;
		if(i != toyear) tomonth_y = 12;
		int frommonth_y = frommonth;
		if(fromyear != i)       frommonth_y = 1;	
        	for(int j = frommonth_y; j <= tomonth_y; j++){
        		int lastday = getMonthDay(i, j);
        		int firstday = 1;
        		if(i == fromyear && j == frommonth)
        			firstday = fromday;
        		if(i == toyear && j == tomonth)
        			lastday = today;
        		String month = j < 10 ? ("0" + j) : ("" + j);
        		for(int k = firstday; k <= lastday; k++){
        			reducernum++;
        			String day = k < 10 ? ("0" + k) : ("" + k);
        			Path p2 = new Path(args[1] + "/" + i + month + "/" + i + month + day);
        			MultipleInputs.addInputPath(job, p2, TextInputFormat.class, Mappersns.class);
        		}
        	}
        }
        Path p3 = new Path(args[4]);
        FileOutputFormat.setOutputPath(job, p3);
        
        job.setReducerClass(Reducer.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);
        
        if(RunChainJob2.isBatchCaids)	job.setNumReduceTasks(RunChainJob2.step2ReducerNum);
	else	job.setNumReduceTasks(1 + reducernum/10);

        return job.waitForCompletion(true) ? 1 : 0;
    }
    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new Configuration(), new Method2CookieUVStep2(), args);
        System.exit(res);
    }
}
